package com.java.basis.thread.n8;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 自定义线程池
 *
 * @author wangning
 * @create 2022-01-30 11:22
 */
@Slf4j(topic = "c.CustomThreadPoolTest01")
public class CustomThreadPoolTest01 {
	public static void main(String[] args) {
		ThreadPool threadPool = new ThreadPool(1, 1000, TimeUnit.MILLISECONDS, 1, (queue, task) -> {
			//死等策略
//			queue.put(task);
			//2.带超时时间
//			queue.offer(task, 1500, TimeUnit.MILLISECONDS);
			//3.抛出异常
//			throw new RuntimeException("任务执行失败" + task);
			//4.让调用者自己执行任务
			task.run();

		});


		for (int i = 0; i < 3; i++) {
			int j = i;
			threadPool.execute(() -> {
				try {
					Thread.sleep(1000L);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}

				log.debug("{}", j);
			});
		}
	}
}

@FunctionalInterface //决绝策略
interface RejectPolicy<T> {
	void reject(BlockQueue<T> queue, T task);

}

@Slf4j(topic = "c.ThreadPool")
class ThreadPool {
	//任务队列
	private BlockQueue<Runnable> taskQueue;
	//线程集合
	private HashSet<Worker> workers = new HashSet<>();
	//核心线程数
	private int coreSize;
	//获取任务的超时时间
	private long timeout;
	//时间单位
	private TimeUnit timeUnit;

	private RejectPolicy<Runnable> rejectPolicy;

	//执行任务
	public void execute(Runnable task) {
		//当任务数没有超过coreSize时，直接交给worker对象执行
		//如果任务数超过了coreSize时，加入任务队列缓存
		synchronized (workers) {
			if (workers.size() < coreSize) {
				Worker worker = new Worker(task);
				log.debug("新增worker {},{}", worker, task);
				workers.add(worker);
				worker.start();
			} else {
//				taskQueue.put(task);//死等
				//1).死等
				//2).带超时时间的等待
				//3).让调用者放弃任务执行
				//4).让调用者抛出异常
				//5).让调用者自己执行
				taskQueue.tryPut(rejectPolicy, task);

			}
		}
	}

	public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity, RejectPolicy<Runnable> rejectPolicy) {
		this.coreSize = coreSize;
		this.timeout = timeout;
		this.timeUnit = timeUnit;
		this.taskQueue = new BlockQueue<>(queueCapacity);
		this.rejectPolicy = rejectPolicy;
	}

	class Worker extends Thread {
		private Runnable task;
		public Worker(Runnable task) {
			this.task = task;
		}

		@Override
		public void run() {
			//执行任务
			//1)当task不为空，执行任务
			//2)当task执行完毕，在接着从任务队列中获取任务并执行
			while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
				try {
					log.debug("正在执行{}", task);
					task.run();
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					task = null;
				}
			}
			synchronized (workers) {
				log.debug("worker被移除 {}", this);
				workers.remove(this);
			}
		}
	}
}

@Slf4j(topic = "c.BlockQueue")
class BlockQueue<T> {
	//1.任务队列
	private Deque<T> queue = new ArrayDeque<>();
	//2.锁
	private ReentrantLock lock = new ReentrantLock();
	//3.生产者条件变量
	private Condition fullWaitSet = lock.newCondition();
	//4.消费者条件变量
	private Condition emptyWaitSet = lock.newCondition();

	//5.容量
	private int capacity;

	public BlockQueue(int capacity) {
		this.capacity = capacity;
	}

	//6.阻塞获取
	public T take() {
		lock.lock();
		try {
			while (queue.isEmpty()) {
				try {
					emptyWaitSet.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T t = queue.removeFirst();
			fullWaitSet.signal();
			return t;

		} finally {
			lock.unlock();
		}
	}

	//7.阻塞添加
	public void put(T task) {
		lock.lock();
		try {
			while (queue.size() == capacity) {
				try {
					log.debug("等待加入任务队列 {}", task);
					fullWaitSet.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			log.debug("加入任务队列 {}", task);
			queue.addLast(task);
			emptyWaitSet.signal();

		} finally {
			lock.unlock();
		}
	}

	//带超时时间的阻塞添加
	public boolean offer(T task, long timeout, TimeUnit timeUnit) {
		lock.lock();
		try {
			long nanos = timeUnit.toNanos(timeout);
			while (queue.size() == capacity) {
				try {
					if (nanos <= 0) {
						return false;
					}
					log.debug("等待加入任务队列 {}", task);
					nanos = fullWaitSet.awaitNanos(nanos);

				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			log.debug("加入任务队列 {}", task);
			queue.addLast(task);
			emptyWaitSet.signal();

		} finally {
			lock.unlock();
		}
		return true;
	}

	//8.获取大小
	public int size() {
		lock.lock();
		try {
			return queue.size();
		} finally {
			lock.unlock();
		}
	}

	//带超时的阻塞获取
	public T poll(long timeout, TimeUnit unit) {
		lock.lock();
		try {
			//将超时时间timeout转化为nanos
			long nanos = unit.toNanos(timeout);
			while (queue.isEmpty()) {
				try {
					//awaitNanos方法返回值时间等于等待剩余的时间
					if (nanos <= 0) {
						return null;
					}
					nanos = emptyWaitSet.awaitNanos(nanos);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T t = queue.removeFirst();
			fullWaitSet.signal();
			return t;
		} finally {
			lock.unlock();
		}
	}

	public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
		lock.lock();
		try {
			//判断队列是否已满
			if (queue.size() == capacity) {
				rejectPolicy.reject(this, task);
			} else {//有空闲
				log.debug("加入任务队列 {}", task);
				queue.addLast(task);
				emptyWaitSet.signal();
			}
		} finally {
			lock.unlock();
		}
	}
}
